package etcd

import (
	"context"
	"fmt"
	"github.com/pkg/errors"
	"github.com/rpcxio/libkv/store"
	"go.etcd.io/etcd/api/v3/mvccpb"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.uber.org/zap"
	"math"
	"sync"
	"time"
	"wzgames/pkg/logger"
)

const (
	DefaultTimeout        = 10 * time.Second
	DefaultRequestTimeout = 10 * time.Second
	DefaultStoreTimeout   = 10 * time.Second

	endID = uint64(math.MaxUint64)
)

// ETCD封装

type EtcdClient struct {
	cfg *EtcdConfig

	options   clientv3.Config
	rawClient *clientv3.Client
	lease     clientv3.Lease

	watcher  clientv3.Watcher
	watchers map[string]*EtcdWatcher

	leaseID       clientv3.LeaseID
	onLeaseChange func(clientv3.LeaseID) error
	ctx           context.Context
	cancel        context.CancelFunc
	timeout       time.Duration

	rw sync.RWMutex

	done chan struct{}

	ttl int64
}

type EtcdWatcher struct {
	cancel context.CancelFunc
}

func (w *EtcdWatcher) Cancel() {
	w.cancel()
}

func NewEtcdClient(cfg *EtcdConfig) *EtcdClient {
	ctx, cancel := context.WithCancel(context.Background())
	if cfg.ConnectionTimeout == 0 {
		cfg.ConnectionTimeout = DefaultTimeout
	}
	config := clientv3.Config{
		Endpoints:          cfg.Addrs,
		DialTimeout:        cfg.ConnectionTimeout,
		MaxCallRecvMsgSize: math.MaxInt32,
		Context:            ctx,
		AutoSyncInterval:   cfg.ConnectionTimeout * 3,
	}
	if cfg.UserName != "" {
		config.Username = cfg.UserName
	}
	if cfg.Password != "" {
		config.Password = cfg.Password
	}
	return &EtcdClient{
		cfg:      cfg,
		options:  config,
		ctx:      ctx,
		cancel:   cancel,
		done:     make(chan struct{}),
		timeout:  cfg.ConnectionTimeout,
		watchers: make(map[string]*EtcdWatcher),
		ttl:      600,
	}
}

// make sure client inited
func (p *EtcdClient) CheckClientKV() error {
	if p.rawClient == nil {
		p.rw.Lock()
		if p.rawClient == nil {
			kv, err := clientv3.New(p.options)
			if err != nil {
				p.cancel()
				logger.Error("[Etcd] cannot create etcd client: ", zap.Error(err), zap.Stack("v3.new"))
				p.rw.Unlock()
				return err
			}
			p.rawClient = kv
			p.lease = clientv3.NewLease(p.rawClient)
			p.watcher = clientv3.NewWatcher(p.rawClient)

			leaseRet, ex := p.Grant(p.ttl + 2)
			if ex != nil {
				panic("[Etcd] create lease fail." + ex.Error())
			}
			p.leaseID = leaseRet.ID
			go p.keepLeaseAlive()
		}
		p.rw.Unlock()
	}

	return nil
}

func (p *EtcdClient) Context() context.Context {
	return p.rawClient.Ctx()
}

func (p *EtcdClient) RawClient() *clientv3.Client {
	return p.rawClient
}

func (p *EtcdClient) NewKV() clientv3.KV {
	return clientv3.NewKV(p.rawClient)
}

func (p *EtcdClient) Close() error {
	if p.rawClient == nil {
		return nil
	}
	for key, watcher := range p.watchers {
		watcher.Cancel()
		delete(p.watchers, key)
	}
	close(p.done)
	p.cancel()

	return p.rawClient.Close()
}

func (p *EtcdClient) Get(key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
	ctx, cancel := context.WithTimeout(p.Context(), DefaultRequestTimeout)
	defer cancel()

	return p.rawClient.Get(ctx, key, opts...)
}

func (p *EtcdClient) GetRawValue(key string, opts ...clientv3.OpOption) ([]byte, error) {
	ctx, cancel := context.WithTimeout(p.Context(), p.timeout)
	resp, err := p.rawClient.Get(ctx, key, opts...)
	cancel()
	if err != nil {
		return nil, err
	}
	if len(resp.Kvs) == 0 {
		return nil, store.ErrKeyNotFound
	}

	return resp.Kvs[0].Value, nil
}

func (p *EtcdClient) GetKey(prefix string, id uint64) string {
	return fmt.Sprintf("%s/%020d", prefix, id)
}

func (p *EtcdClient) GetKeyNopadding(prefix string, id uint64) string {
	return fmt.Sprintf("%s/%d", prefix, id)
}

func (p *EtcdClient) GetRawValues(dir string, fn func(string, []byte) error) error {
	resp, err := p.Get(dir, clientv3.WithPrefix())
	if err != nil {
		return err
	}

	if len(resp.Kvs) == 0 {
		return nil
	}

	for _, item := range resp.Kvs {
		err = fn(string(item.Key), item.Value)
		if err != nil {
			logger.Error("[ETCD.kv] GetValues Error", zap.String("Dir", dir), zap.Error(err))
			return err
		}
	}

	return nil
}

func (p *EtcdClient) Watch(ctx context.Context, key string, prefix bool, watchEventC chan *KvEvt) error {
	p.rw.Lock()
	if _, ok := p.watchers[key]; ok {
		p.rw.Unlock()
		return errors.Errorf("watch prefix %s already registered", key)
	}
	nctx, cancel := context.WithCancel(ctx)
	p.watchers[key] = &EtcdWatcher{cancel: cancel}
	p.rw.Unlock()

	var watchChan clientv3.WatchChan
	if prefix {
		watchChan = p.watcher.Watch(nctx, key, clientv3.WithPrefix(), clientv3.WithPrevKV())
	} else {
		watchChan = p.watcher.Watch(nctx, key, clientv3.WithPrevKV())
	}

	go func() {
		for {
			select {
			case <-nctx.Done():
				goto EXIT
			case wresp := <-watchChan:
				err := wresp.Err()
				if err != nil {
					goto EXIT
				}
				if wresp.Canceled {
					goto EXIT
				}
				for _, ev := range wresp.Events {
					var evtSrc KvEvtSrc
					var evtType KvEvtType
					switch ev.Type {
					case mvccpb.DELETE:
						evtType = KvEventDelete
					case mvccpb.PUT:
						if ev.IsCreate() {
							evtType = KvEventNew
						} else if ev.IsModify() {
							evtType = KvEventUpdate
						}
					}
					watchEventC <- &KvEvt{
						Src:   evtSrc,
						Type:  evtType,
						Key:   string(ev.Kv.Key),
						Value: ev.Kv.Value,
					}
				}
			}
		EXIT:
		}
	}()

	return nil
}

func (p *EtcdClient) UnWatch(key string) {
	p.rw.Lock()
	defer p.rw.Unlock()
	if watcher, ok := p.watchers[key]; ok {
		watcher.Cancel()
		delete(p.watchers, key)
	}
}

func (p *EtcdClient) Store(key, value string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
	ctx, cancel := context.WithTimeout(p.Context(), DefaultStoreTimeout)
	defer cancel()
	return p.rawClient.Put(ctx, key, value, opts...)
}

func (p *EtcdClient) StoreWithModRev(key, value string, rev int64) (*clientv3.PutResponse, error) {
	if rev == 0 {
		return p.Store(key, value)
	}
	ctx, cancel := context.WithTimeout(p.Context(), DefaultStoreTimeout)
	tresp, err := p.rawClient.Txn(ctx).
		If(clientv3.Compare(clientv3.ModRevision(key), "=", rev)).
		Then(clientv3.OpPut(key, value)).
		Commit()
	cancel()
	if err != nil {
		return nil, err
	}

	if !tresp.Succeeded {
		return nil, ErrValueMayChanged
	}

	resp := clientv3.PutResponse(*tresp.Responses[0].GetResponsePut())
	return &resp, nil
}

func (p *EtcdClient) StoreEphemeral(key, value string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
	ctx, cancel := context.WithTimeout(p.Context(), DefaultStoreTimeout)
	opts = append(opts, clientv3.WithLease(p.leaseID))
	tresp, err := p.rawClient.Txn(ctx).
		Then(clientv3.OpPut(key, value, opts...)).
		Commit()
	cancel()

	if err != nil {
		return nil, err
	}

	if !tresp.Succeeded {
		logger.Error("[Etcd] Failed to store in txn", zap.String("key", key))
		return nil, fmt.Errorf("failed to store %s in txn", key)
	}

	resp := clientv3.PutResponse(*tresp.Responses[0].GetResponsePut())

	return &resp, nil
}

func (p *EtcdClient) Delete(key string, opts ...clientv3.OpOption) (*clientv3.DeleteResponse, error) {
	ctx, cancel := context.WithTimeout(p.Context(), DefaultStoreTimeout)
	defer cancel()

	return p.rawClient.Delete(ctx, key, opts...)
}

func (p *EtcdClient) GetLock(key string, id clientv3.LeaseID) (bool, error) {
	ctx, cancel := context.WithTimeout(p.Context(), DefaultStoreTimeout)
	resp, err := p.rawClient.Txn(ctx).
		If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
		Then(clientv3.OpPut(key, "", clientv3.WithLease(id))).
		Else(clientv3.OpGet(key)).
		Commit()
	cancel()

	if err != nil {
		return false, err
	}

	return resp.Succeeded, nil
}

func (p *EtcdClient) DelLock(key string) error {
	_, err := p.Delete(key)
	return err
}

func (p *EtcdClient) OnLeaseChanged(fn func(clientv3.LeaseID) error) {
	p.onLeaseChange = fn
}

func (p *EtcdClient) Grant(ttl int64) (*clientv3.LeaseGrantResponse, error) {
	_ctx, cancel := context.WithTimeout(p.Context(), DefaultRequestTimeout)
	defer cancel()

	return p.lease.Grant(_ctx, ttl)
}

func (p *EtcdClient) Revoke(id clientv3.LeaseID) (*clientv3.LeaseRevokeResponse, error) {
	ctx, cancel := context.WithTimeout(p.Context(), DefaultRequestTimeout)
	defer cancel()
	return p.lease.Revoke(ctx, id)
}

func (p *EtcdClient) KeepAliveOnce(id clientv3.LeaseID) (*clientv3.LeaseKeepAliveResponse, error) {
	_ctx, cancel := context.WithTimeout(p.Context(), DefaultRequestTimeout)
	defer cancel()
	return p.lease.KeepAliveOnce(_ctx, id)
}

func (p *EtcdClient) keepLeaseAlive1(leaseID clientv3.LeaseID) {
	keepRespChan, err := p.lease.KeepAlive(p.Context(), leaseID)
	if err != nil {
		logger.Error("[Etcd] lease keep alive Error", zap.Error(err))
		return
	}

	for {
		select {
		case keepResp := <-keepRespChan:
			if keepResp == nil {
				logger.Warn("[Etcd] lease id expire", zap.Any("LeaseID", leaseID))
				return
			} else {
				//plog.Info("[Etcd] receive lease response", log.Object("LeaseID", keepResp.ID))
			}

		}
	}
}

func (p *EtcdClient) keepLeaseAlive() {
	duration := time.Duration(p.ttl) * time.Second
	timer := time.NewTimer(duration)
	for {
		select {
		case <-p.done:
			return
		case <-timer.C:
			if p.ttl == 0 {
				return
			}
			if p.LeaseID() > 0 {
				_, err := p.KeepAliveOnce(p.LeaseID())
				if err == nil {
					timer.Reset(duration)
					continue
				}

				logger.Warn("[Etcd] keepAlive error, try to reset...", zap.Int64("LeaseID", int64(p.LeaseID())), zap.Error(err))
				p.leaseID = 0
			}

			lid, ex := p.Grant(p.ttl + 2)
			if ex != nil {
				logger.Warn("[Etcd] set leaseID error, try to reset...", zap.Error(ex))
			} else {
				p.leaseID = lid.ID
				if p.onLeaseChange != nil {
					_ = p.onLeaseChange(lid.ID)
				}
				logger.Info("[Etcd] set leaseID success", zap.Int64("LeaseID", int64(p.LeaseID())))
			}
			timer.Reset(duration)
		}
	}
}

func (p *EtcdClient) LeaseID() clientv3.LeaseID {
	return p.leaseID
}
